package com.atguigu.wuliu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.wuliu.app.func.BeanToJsonStrFunction;
import com.atguigu.wuliu.app.func.DimAsyncByForeignKeyFunction;
import com.atguigu.wuliu.bean.DWSTradeOrgOrderBean;
import com.atguigu.wuliu.utils.DateFormatUtil;
import com.atguigu.wuliu.utils.DorisUtil;
import com.atguigu.wuliu.utils.KafkaUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.util.concurrent.TimeUnit;

public class DwsTradeOrgOrderDay {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        env.enableCheckpointing(5000L);

    String topic = "tms_dwd_trade_order_detail";
    String groupId = "dws_trade_org_order_group";
        KafkaSource<String> kafkaSource = KafkaUtil.getKafkaSource(topic, groupId);
        DataStreamSource<String> kafkaStrDS = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka_source");
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String jsonStr, Context ctx, Collector<JSONObject> out) throws Exception {
                        if (StringUtils.isNotEmpty(jsonStr)) {
                            JSONObject jsonObj = JSON.parseObject(jsonStr);
                            out.collect(jsonObj);
                        }
                    }
                }
        );
//jsonObjDS.print("<<<");
        /*{"cargoType":"74002","amount":13.00,"orderNo":"2023-1079454639-00000818",
        "distance":849.00,"estimateArriveTime":"2023-08-21 13:00:31","orderId":"29286",
        "receiverName":"韩*","receiverCityId":"110100","weight":4.00,"receiverProvinceId":"110000",
        "userId":"73","receiverComplexId":"92","cargoNum":2,"collectType":"69002",
        "senderComplexId":"79","senderName":"孟*","orderTime":"2023-08-18 12:00:31",
        "receiverDistrictId":"110109","senderDistrictId":"140107","senderProvinceId":"140000",
        "senderCityId":"140100","id":"29284","status":"60010","ts":1692331231000}*/
        SingleOutputStreamOperator<DWSTradeOrgOrderBean> orderBeanDS  = jsonObjDS.map(
                new MapFunction<JSONObject, DWSTradeOrgOrderBean>() {
                    @Override
                    public DWSTradeOrgOrderBean map(JSONObject jsonObj) throws Exception {
                        String amount = jsonObj.getString("amount");
                        Long ts = jsonObj.getLong("ts") ;
                        String senderDistrictId = jsonObj.getString("senderDistrictId");
                        String senderCityId = jsonObj.getString("senderCityId");
                        return DWSTradeOrgOrderBean.builder()
                                .orderCountBase(1L)
                                .orderAmountBase(new BigDecimal(amount))
                                .senderDistrictId(senderDistrictId)
                                .cityId(senderCityId)
                                .ts(ts)
                                .build();
                    }
                }

        );
//orderBeanDS.print("////");
//DWSTradeOrgOrderBean(stt=null, edt=null, curDate=null, orgId=null, orgName=null, cityId=220100, cityName=null, senderDistrictId=220104, orderAmountBase=11.00, orderCountBase=1, ts=1692318601000000)

        //5.关联机构数据：通过地区id关联维度表：dim_base_organ 中的 id，获取机构名：org_name
        SingleOutputStreamOperator<DWSTradeOrgOrderBean> joinDS = AsyncDataStream.unorderedWait(
                orderBeanDS,
                new DimAsyncByForeignKeyFunction<DWSTradeOrgOrderBean>("dim_base_organ") {
                    @Override
                    public void join(DWSTradeOrgOrderBean obj, JSONObject dimInfoJsonObj) {
                        obj.setOrgId(dimInfoJsonObj.getString("id"));
                        obj.setOrgName(dimInfoJsonObj.getString("org_name"));
                    }

                    @Override
                    public Tuple2<String, String> getForeignKey(DWSTradeOrgOrderBean obj) {
                        return Tuple2.of("region_id", obj.getSenderDistrictId());
                    }
                }, 10, TimeUnit.SECONDS
        );

//joinDS.print("---");


        SingleOutputStreamOperator<DWSTradeOrgOrderBean> withWatermarkDS  = joinDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<DWSTradeOrgOrderBean>forMonotonousTimestamps().withTimestampAssigner(
                        new SerializableTimestampAssigner<DWSTradeOrgOrderBean>() {
                            @Override
                            public long extractTimestamp(DWSTradeOrgOrderBean orderBean, long recordTimestamp) {
                                return orderBean.getTs();
                            }
                        }
                )
        );


        KeyedStream<DWSTradeOrgOrderBean, String> provinceIdKeyedDS
                = withWatermarkDS.keyBy(DWSTradeOrgOrderBean::getOrgId);

        WindowedStream<DWSTradeOrgOrderBean, String, TimeWindow> windowDS
                = provinceIdKeyedDS.window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10)));


        SingleOutputStreamOperator<DWSTradeOrgOrderBean> reduceDS = windowDS.reduce(new ReduceFunction<DWSTradeOrgOrderBean>() {
                                                                                      @Override
                                                                                      public DWSTradeOrgOrderBean reduce(DWSTradeOrgOrderBean value1, DWSTradeOrgOrderBean value2) throws Exception {
                                                                                          value1.setOrderAmountBase(value1.getOrderAmountBase().add(value2.getOrderAmountBase()));
                                                                                          value1.setOrderCountBase(value1.getOrderCountBase() + value2.getOrderCountBase());
                                                                                          return value1;
                                                                                      }
                                                                                  },
                new WindowFunction<DWSTradeOrgOrderBean, DWSTradeOrgOrderBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<DWSTradeOrgOrderBean> input, Collector<DWSTradeOrgOrderBean> out) throws Exception {
                        String stt = DateFormatUtil.toYmdHms(window.getStart());
                        String edt = DateFormatUtil.toYmdHms(window.getEnd());
                        String curDate = DateFormatUtil.toPartitionDate(window.getStart());
                        for (DWSTradeOrgOrderBean orderBean : input) {
                            orderBean.setStt(stt);
                            orderBean.setEdt(edt);
                            orderBean.setCurDate(curDate);
                            out.collect(orderBean);
                        }
                    }
                }
        );
        //reduceDS.print("****");

        SingleOutputStreamOperator<DWSTradeOrgOrderBean> withcitynameDS = AsyncDataStream.unorderedWait(
                reduceDS,
                new DimAsyncByForeignKeyFunction<DWSTradeOrgOrderBean>("dim_base_region_info") {
                    @Override
                    public void join(DWSTradeOrgOrderBean obj, JSONObject dimInfoJsonObj) {
                        obj.setCityName(dimInfoJsonObj.getString("name"));
                    }

                    @Override
                    public Tuple2<String, String> getForeignKey(DWSTradeOrgOrderBean obj) {
                        return Tuple2.of("id", obj.getCityId());
                    }
                }, 10, TimeUnit.SECONDS
        );

withcitynameDS.print("555");


        withcitynameDS.map(
                new BeanToJsonStrFunction<>()
        ).sinkTo(DorisUtil.getDorisSink("dws_trade_org_order_day_base"));
        env.execute();

    }

}
